package com.sunzm.spark.sql.exercise

import org.apache.commons.lang3.StringUtils
import org.apache.spark.sql.SparkSession

/**
 *
 * ${DESCRIPTION}
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-06-07 18:57
 */
object SparkSQLFeeJob2 {

  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName(s"${this.getClass.getSimpleName.stripSuffix("$")}")
      .master("local[*]")
      .config("spark.default.parallelism", 6)
      .config("spark.sql.shuffle.partitions", 6)
      .getOrCreate()

    //这里读取的是整月的全量数据
    val dataDS = spark.read.textFile("G:\\data\\offlinetask\\eSDataTmpExport\\report\\callLogData\\callData-2021-04-01-2021-05-01.csv")

    import spark.implicits._

    dataDS.filter(line => {
      var res = false
      if (!StringUtils.contains(line, "主通话ID")) {
        val fields = StringUtils.splitByWholeSeparatorPreserveAllTokens(line, ",")
        val callDirectionStr = StringUtils.trim(StringUtils.replace(fields(9), "\t", ""))
        val feeDuration = StringUtils.trim(StringUtils.replace(fields(24), "\t", "")).toLong
        val fee = StringUtils.trim(StringUtils.replace(fields(25), "\t", "")).toDouble
        if (!StringUtils.contains(callDirectionStr, "呼出") && (fee > 0 || feeDuration > 0)) {
          res = true
        }
      }
      res
    }).show(10)

    val mapDS = dataDS.filter(line => {
      var res = false
      if (!StringUtils.contains(line, "主通话ID")) {
        val fields = StringUtils.splitByWholeSeparatorPreserveAllTokens(line, ",")
        val callDirectionStr = StringUtils.trim(StringUtils.replace(fields(9), "\t", ""))
        val fee = StringUtils.trim(StringUtils.replace(fields(25), "\t", "")).toDouble
        if (StringUtils.contains(callDirectionStr, "呼出") && fee > 0) {
          res = true
        }
      }
      res
    }).map(line => {
      val fields = StringUtils.splitByWholeSeparatorPreserveAllTokens(line, ",")
      val caller = StringUtils.trim(StringUtils.replace(fields(13), "\t", ""))
      //计费时长, 单位：分钟
      val feeDuration = StringUtils.trim(StringUtils.replace(fields(24), "\t", "")).toLong
      //费用，转为分，计算比较精确
      val fee = (StringUtils.trim(StringUtils.replace(fields(25), "\t", "")).toDouble * 100).toLong

      (caller, feeDuration, fee)
    }).toDF( "caller", "feeDuration", "fee")
      .as[(String, Long, Long)]

    val sum1: Long = mapDS.filter(t => {
      val caller = t._1
      val trimCaller = StringUtils.trim(caller)
      val callerLen = StringUtils.length(trimCaller)

      callerLen == 11 && StringUtils.startsWith(trimCaller, "1")
    }).map(_._2)
      .reduce(_ + _)
    //.show(1000)

    println(s"手机号：${sum1}")

    val sum2: Long = mapDS.filter(t => {
      val caller = t._1
      val trimCaller = StringUtils.trim(caller)
      val callerLen = StringUtils.length(trimCaller)

      !(callerLen == 11 && StringUtils.startsWith(trimCaller, "1"))
    }).map(_._2)
      .reduce(_ + _)

    println(s"固话：${sum2}")

    println(s"合计：${sum1} + ${sum2} = ${sum1 + sum2}")

    spark.close()
  }

}
